package cn.study.base;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class LongEventMain {

	public static void main(String[] args) throws Exception {
		// 创建缓冲线程池
		ExecutorService  executor = Executors.newCachedThreadPool();
		// 创建工厂
		LongEventFactory factory = new LongEventFactory();
		/**
		 * 创建bufferSize ,也就是RingBuffer大小，必须是2的N次方
		 *
		 * 2的N次方换成二进制就是1000，100，10，1这样的数字，
		 * sequence & （array length－1） = array index，
		 * 比如一共有8槽，3&（8－1）=3，HashMap就是用这个方式来定位数组元素的，
		 * 这种方式比取模的速度更快。
 		 */
		int ringBufferSize = 1024 * 1024; //

		/**

		*/
		
		// 创建disruptor
		/**
		 * 参数一：工厂类对象，用于常见一个个的LongEvent,LongEvent就是实际的消费数据
		 * 参数二：缓冲区大小
		 * 参数三：线程池，进行disruptor内部的数据接收处理调度
		 * 参数四：ProducerType.SINGLE表示有一个生产者 ProducerType.MULTI表示有多个生产者
		 * 参数五：WaitStrategy策略
		 * 		BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
		 * 			WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();
		 * 		SleepingWaitStrategy 的性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景
		 * 			WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();
		 * 		YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性
		 * 			WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
		 */
		Disruptor<LongEvent> disruptor = 
				new Disruptor<LongEvent>(factory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
		// 连接消费事件方法，处理数据
		disruptor.handleEventsWith(new LongEventHandler());
		
		// 启动disruptor
		disruptor.start();
		
		// Disruptor 的事件发布过程是一个两阶段提交的过程：
		// 发布事件（获取存放数据的容器）
		RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
		// 实例化生产者，并将RingBuffer容器注入到生产者
		// LongEventProducer producer = new LongEventProducer(ringBuffer);
		LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);

		ByteBuffer byteBuffer = ByteBuffer.allocate(8);
		for(long l = 0; l<100; l++) {
			byteBuffer.putLong(0, l);
			producer.onData(byteBuffer);
			// Thread.sleep(1000);
		}

		disruptor.shutdown();// 关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
		executor.shutdown();// 关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；
	}
}
